package com.neofinance.reactdemo;

import lombok.extern.java.Log;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.client.WebClient;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by Michael Chen on 2017-8-22.
 */
@Log
public class ReactiveClientApplication {

    private static final ExecutorService executorService = Executors.newCachedThreadPool();

    /**
     * 设置SLF4J日志输出级别
     *
     * @param level 日志级别
     */
    public static void setLoggingLevel(ch.qos.logback.classic.Level level) {
        ch.qos.logback.classic.Logger root = (ch.qos.logback.classic.Logger) org.slf4j.LoggerFactory.getLogger(ch.qos.logback.classic.Logger.ROOT_LOGGER_NAME);
        root.setLevel(level);
    }

    private static class Task implements Runnable {
        @Override
        public void run() {
            WebClient.create("http://shenzhenryan.asuscomm.com:8090/")
                    .get()
                    .uri("/events")
                    .accept(MediaType.TEXT_EVENT_STREAM)
                    .exchange()
                    .flatMapMany(cr -> cr.bodyToFlux(Event.class))
                    .subscribe(event -> log.info("Thread ID:【" + Thread.currentThread().getId() + "】 " + Thread.currentThread().getName() + "::::::::" + event.toString()));
        }
    }

    public static void main(String[] args) throws Exception {
        // 屏蔽WebClient引入的SLF4J调试日志
        setLoggingLevel(ch.qos.logback.classic.Level.INFO);
        // 启动5个客户端线程
        for (int i = 0; i < 5; i++) {
            executorService.submit(new Task());
        }
        log.info("系统时间："+System.currentTimeMillis());
    }
}
